package knowlwdge_graph.knowledge_etl.yago_etl;

import common.dao.postgre.PostGreJDBCDao;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by julianzliu on 4/15/2017.
 */
public class ExtractorPostgres {

    static final org.slf4j.Logger logger = LoggerFactory.getLogger(ExtractorPostgres.class);

    private PostGreJDBCDao postGreJDBCDao;
    private Connection conn;

    public ExtractorPostgres(){
        this.postGreJDBCDao = new PostGreJDBCDao();

        this.conn = this.postGreJDBCDao.getConn();

        if(this.conn == null){
            logger.error("[ExtractorPostgres] Connection Failed!!!\n\n");
        }
    }

    public List<RowYagoTriple> queryAllColsBySQL(String sql){
        Statement stmt=null;
        ResultSet rs=null;
        List<RowYagoTriple> rows = new ArrayList<>();
        try
        {
            stmt=this.conn.createStatement();
            rs=stmt.executeQuery(sql);
            while(rs.next()){
                String id = rs.getString(1);
                if (id!=null) id = id.replace("'","\\'");
                Float value = rs.getFloat(5);
                if(value==null) value = 0.0f;
                rows.add(
                        new RowYagoTriple(
                                id,//("id"),
                                rs.getString(2).replace("'","\\'"),//("subject"),
                                rs.getString(3).replace("'","\\'"),//("predicate"),
                                rs.getString(4).replace("'","\\'"),//("object"),
                                value//("value")
                        )
                );
            }

            rs.close();
            stmt.close();
            //conn.close();
        }
        catch (SQLException e)
        {
            e.printStackTrace();
        }
        return rows;
    }

    public List<RowYagoTriple> getPartitialRowSet(Integer offset, Integer limit){

        String sql = String.format("select * from yagofacts order by id offset %d limit %d;", offset, limit );

        logger.info("[getPartitialRowSet] sql = \n{}", sql);

        return queryAllColsBySQL(sql);

    }

    /*
    每次查询结束之后，在数据库中保存offset
    表结构：
    CREATE TABLE yagostatus (
      ts timestamp default current_timestamp,
      id integer not null,
      status integer not null
    )
     */
    public boolean storeLatestOffset(Integer offset, Integer status){
        String sql = String.format("insert into yagostatus (id,status) values (%d,%d);", offset,status);
        logger.info("[storeLatestOffset] sql = {}", sql);
        boolean success = true;
        try {
            Statement stmt = conn.createStatement();
            success = stmt.execute(sql);
            stmt.close();
        } catch (SQLException e){
            e.printStackTrace();
            success = false;
        }
        return success;
    }

    public Integer getLatestOffset(){
        String sql = String.format("select * from yagostatus where status = 1 order by ts desc limit 1;");
        logger.info("[getLatestOffset] sql = {}", sql);
        Integer offset = 0;
        try {
            Statement stmt = conn.createStatement();
            ResultSet rs = null;
            rs = stmt.executeQuery(sql);
            while(rs.next()){
                offset = rs.getInt("id");
            }
            rs.close();
            stmt.close();
        } catch (SQLException e){
            e.printStackTrace();
        }
        return offset;
    }


    /******************************************************************
     * 查询主语或谓语是某个实体的所有三元组，导入OrientDB
     */
    public List<RowYagoTriple> getAllTripeOfEntity(String yago_id){
        String sql = String.format("select * from yagofacts where subject = '%s' or object = '%s';", yago_id, yago_id );

        logger.info("[getAllTripeOfEntity] sql = \n{}", sql);

        return queryAllColsBySQL(sql);
    }






    /******************************************************************
     * MAIN
     */
    public static void main(String[] args){

        ExtractorPostgres extractor = new ExtractorPostgres();
        Integer offset = 100;
        Integer limit = 100;
        List<RowYagoTriple> rows = extractor.getPartitialRowSet(offset, limit);
        for(RowYagoTriple row: rows){
            System.out.println(row.id + " " + row.subject + " " + row.predicate + " " + row.object + " " + row.value);
        }

        offset += limit;
        extractor.storeLatestOffset(offset, 1);
        Integer latest_offset = extractor.getLatestOffset();
        System.out.println("latest offset = " + latest_offset);
    }

}
